package mq

import (
	"context"
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/streadway/amqp"
)

// RabbitMQPlugin 实现基于 RabbitMQ 的消息队列
type RabbitMQPlugin struct {
	connString      string
	closed          bool
	startTime       int64
	errorCount      int
	lastError       string
	queueStats      map[string]*QueueStats
	statsMutex      sync.RWMutex
	consumersMutex  sync.Mutex
	activeConsumers map[string]*amqp.Channel
}

// 确保 RabbitMQPlugin 实现了 MQPlugin 接口
var _ MQPlugin = (*RabbitMQPlugin)(nil)

// newRabbitMQPlugin 创建一个新的 RabbitMQPlugin 实例
func newRabbitMQPlugin(config map[string]interface{}) (MQPlugin, error) {
	// 从配置中获取连接字符串
	connString, ok := config["conn_string"].(string)
	if !ok || connString == "" {
		return nil, fmt.Errorf("RabbitMQ 连接字符串配置缺失或无效")
	}

	return &RabbitMQPlugin{
		connString:      connString,
		closed:          false,
		startTime:       time.Now().Unix(),
		queueStats:      make(map[string]*QueueStats),
		activeConsumers: make(map[string]*amqp.Channel),
	}, nil
}

// Type 返回消息队列类型
func (p *RabbitMQPlugin) Type() MQType {
	return RabbitMQ
}

// Publish 发布消息到指定主题
func (p *RabbitMQPlugin) Publish(ctx context.Context, topic string, message string) error {
	conn, err := amqp.Dial(p.connString)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("连接到 RabbitMQ 失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("连接到 RabbitMQ 失败: %w", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("打开通道失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("打开通道失败: %w", err)
	}
	defer ch.Close()

	q, err := ch.QueueDeclare(
		topic, // 队列名称
		true,  // 持久化
		false, // 不使用时删除
		false, // 独占
		false, // 非阻塞
		nil,   // 参数
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("声明队列失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("声明队列失败: %w", err)
	}

	err = ch.Publish(
		"",     // 交换机
		q.Name, // 路由键
		false,  // 强制
		false,  // 立即
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("发布消息失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("发布消息失败: %w", err)
	}

	return nil
}

// Subscribe 订阅指定主题的消息
func (p *RabbitMQPlugin) Subscribe(ctx context.Context, topic string, handler func(message string) error) error {
	conn, err := amqp.Dial(p.connString)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("连接到 RabbitMQ 失败: %v", err)
		log.Printf(p.lastError)
		return fmt.Errorf("连接到 RabbitMQ 失败: %w", err)
	}

	ch, err := conn.Channel()
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("打开通道失败: %v", err)
		log.Printf(p.lastError)
		conn.Close()
		return fmt.Errorf("打开通道失败: %w", err)
	}

	q, err := ch.QueueDeclare(
		topic, // 队列名称
		true,  // 持久化
		false, // 不使用时删除
		false, // 独占
		false, // 非阻塞
		nil,   // 参数
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("声明队列失败: %v", err)
		log.Printf(p.lastError)
		ch.Close()
		conn.Close()
		return fmt.Errorf("声明队列失败: %w", err)
	}

	msgs, err := ch.Consume(
		q.Name, // 队列
		"",     // 消费者
		true,   // 自动确认
		false,  // 独占
		false,  // 非本地
		false,  // 非阻塞
		nil,    // 参数
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("注册消费者失败: %v", err)
		log.Printf(p.lastError)
		ch.Close()
		conn.Close()
		return fmt.Errorf("注册消费者失败: %w", err)
	}

	// 初始化或更新队列统计信息
	p.statsMutex.Lock()
	if _, exists := p.queueStats[topic]; !exists {
		p.queueStats[topic] = &QueueStats{
			TopicName: topic,
		}
	}
	p.statsMutex.Unlock()

	go func() {
		defer ch.Close()
		defer conn.Close()

		for {
			select {
			case <-ctx.Done():
				return
			case msg, ok := <-msgs:
				if !ok {
					return
				}

				// 更新统计信息
				p.statsMutex.Lock()
				stats := p.queueStats[topic]
				stats.LastConsumedAt = time.Now().Unix()
				stats.ConsumedMessages++
				p.statsMutex.Unlock()

				if err := handler(string(msg.Body)); err != nil {
					log.Printf("处理消息失败: %v", err)

					// 更新失败统计
					p.statsMutex.Lock()
					stats.FailedMessages++
					p.statsMutex.Unlock()
				}
			}
		}
	}()

	return nil
}

// Close 关闭 RabbitMQ 连接
func (p *RabbitMQPlugin) Close() error {
	p.closed = true
	return nil
}

// GetStatus 获取RabbitMQ的状态
func (p *RabbitMQPlugin) GetStatus() MQStatus {
	// 尝试建立连接以检查连接状态
	conn, err := amqp.Dial(p.connString)
	isConnected := err == nil
	if isConnected {
		conn.Close() // 关闭测试连接
	} else {
		p.errorCount++
		p.lastError = err.Error()
	}

	return MQStatus{
		IsConnected:    isConnected,
		ConnectionInfo: p.connString, // 可能需要脱敏处理
		StartTime:      p.startTime,
		ErrorCount:     p.errorCount,
		LastError:      p.lastError,
	}
}

// GetQueueStats 获取指定主题的队列统计信息
func (p *RabbitMQPlugin) GetQueueStats(topic string) (QueueStats, error) {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	// 检查队列是否存在
	stats, exists := p.queueStats[topic]
	if !exists {
		return QueueStats{}, fmt.Errorf("主题 %s 不存在或未被监控", topic)
	}

	// 尝试从RabbitMQ获取实时队列信息
	conn, err := amqp.Dial(p.connString)
	if err != nil {
		return *stats, nil // 返回缓存的统计信息
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		return *stats, nil
	}
	defer ch.Close()

	// 获取队列信息
	queue, err := ch.QueueInspect(topic)
	if err != nil {
		return *stats, nil
	}

	// 更新统计信息
	stats.PendingMessages = int64(queue.Messages)

	return *stats, nil
}

// ListQueueStats 获取所有主题的队列统计信息
func (p *RabbitMQPlugin) ListQueueStats() []QueueStats {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	result := make([]QueueStats, 0, len(p.queueStats))
	for _, stats := range p.queueStats {
		result = append(result, *stats)
	}

	return result
}

// GetPendingMessages 获取待处理的消息
func (p *RabbitMQPlugin) GetPendingMessages(topic string, page, pageSize int) ([]string, error) {
	// 实现获取待处理消息的逻辑
	return []string{}, nil
}

// GetConsumedMessages 获取已消费的消息
func (p *RabbitMQPlugin) GetConsumedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取已消费消息的逻辑
	return []MessageRecord{}, nil
}

// GetFailedMessages 获取失败的消息
func (p *RabbitMQPlugin) GetFailedMessages(topic string, page, pageSize int) ([]MessageRecord, error) {
	// 实现获取失败消息的逻辑
	return []MessageRecord{}, nil
}

// CloseQueue 关闭指定主题的队列
func (p *RabbitMQPlugin) CloseQueue(topic string) error {
	// 检查连接状态
	if p.closed {
		return fmt.Errorf("RabbitMQ 连接已关闭")
	}

	// 创建新的连接
	conn, err := amqp.Dial(p.connString)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("连接到 RabbitMQ 失败: %v", err)
		return fmt.Errorf("连接到 RabbitMQ 失败: %w", err)
	}
	defer conn.Close()

	// 从活跃消费者中查找并关闭
	p.consumersMutex.Lock()
	defer p.consumersMutex.Unlock()

	ch, exists := p.activeConsumers[topic]
	if !exists {
		// 如果没有活跃的消费者，创建一个新通道来删除队列
		ch, err = conn.Channel()
		if err != nil {
			p.errorCount++
			p.lastError = fmt.Sprintf("获取 RabbitMQ 通道失败: %v", err)
			return fmt.Errorf("获取 RabbitMQ 通道失败: %w", err)
		}
		defer ch.Close()
	} else {
		// 从活跃消费者映射中移除
		delete(p.activeConsumers, topic)
	}

	// 删除队列
	_, err = ch.QueueDelete(
		topic, // 队列名称
		false, // ifUnused - 如果为true，则仅在队列未使用时删除
		false, // ifEmpty - 如果为true，则仅在队列为空时删除
		false, // noWait - 如果为true，则不等待服务器响应
	)
	if err != nil {
		p.errorCount++
		p.lastError = fmt.Sprintf("删除 RabbitMQ 队列失败: %v", err)
		return fmt.Errorf("删除 RabbitMQ 队列失败: %w", err)
	}

	// 从统计信息中移除
	p.statsMutex.Lock()
	delete(p.queueStats, topic)
	p.statsMutex.Unlock()

	return nil
}

// ListTopics 获取所有主题列表
func (p *RabbitMQPlugin) ListTopics() []string {
	p.statsMutex.RLock()
	defer p.statsMutex.RUnlock()

	topics := make([]string, 0, len(p.queueStats))
	for topic := range p.queueStats {
		topics = append(topics, topic)
	}

	return topics
}
